from tool import logging,getTimeStr,change_task_status
from config import *
from sql import sqlseverDB
from shop import PHShop,Shop
from multiprocessing import Pool, Manager
import traceback,time,sys,random
taskid=int(sys.argv[1])
ms = sqlseverDB('{SQL Server}', SQL_SERVER, 'sumaitongpol', SQL_USER, SQL_PWD)
rs_task=ms.ExecQuerydict('select * from task where id=?',(taskid,))
task_info=rs_task[0]
rw_name=task_info['tName']
isMulti=task_info['isMulti']
last_tid=task_info['LastHouTaiID']

logger = logging.getLogger()
logger.setLevel(logging.INFO) 
log_path=f'D:/Web/Gtupdate/GTLOG/{rw_name}{getTimeStr()}.log'
file_handler = logging.FileHandler(log_path, mode='a', encoding='utf-8')
ms.ExecNoQuery('update task set logPath=? where id=?',(log_path,taskid))
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)


def process_shop(tid,cur_shop,queue):

    try:
        cur_shop.start_task(tid)
    except Exception as e:
        traceback.print_exc()
        time.sleep(3)
    # 将采集后的结果添加到队列
    queue.put(1)


def update_task_status(ms,queue, task_count,tid,hid):
    """更新任务状态"""
    remaining_tasks = task_count - queue.qsize()
    ms.ExecNoQuery('update [task] set [tcount]=?,LastHouTaiID=? where [id]=?', (remaining_tasks,hid,tid))
    rw_name=Task_Name_dict[tid]

    logging.info(f'{rw_name}任务队列剩余: {remaining_tasks}\n')

def on_task_complete(ms,queue, task_count,tid,hid):
    """任务完成时的回调函数"""
    update_task_status(ms,queue, task_count,tid,hid)

def main():
    if taskid in (12,13,14):
        aff_reset=ms.ExecNoQuery(''' update ShopeePHAccount set req_count=0
                        WHERE (last_end_time is null or DATEDIFF(SECOND, DATEADD(SECOND, last_end_time, '1970-01-01'),GETUTCDATE()) > 7200)
                        and req_count!=0  ''')
        print(f'成功将 {aff_reset} 个菲律宾买家号设置为可采集状态')
        sql_str=f'''select * from ShopeePHAccount where req_count<{PH_MAX_REQ_COUNT}  order by req_count'''
    elif taskid in (25,):
        sql_str="select * from houtai where is_publish_bigdata=1 and is_task_baned=0 and dpname not like '%号大店' and dpname not like '%假期%' "
    else:
        sql_str='select * from S1688Account where is_cj_pro=1'

    rs_dp=ms.ExecQuerydict(sql_str)




    if len(rs_dp)==0:
        logging.info(f'任务《{rw_name}》,无账号可以用来采集,等待两小时...')
        ms.ExecNoQuery('update [task] set state=0,tcount=0,LastHouTaiID=null where [id]=?',(taskid,))
        change_task_status(taskid,0)
        return
    
    if taskid in (25,):
        random.shuffle(rs_dp)

        

    if taskid==12:
        sql_lm=''' select * from phlmurl where (lmurl is not null and  lmurl not in ('','原始链接下架')) 
                    and (lmcj_timestamp is null or ABS(DATEDIFF(SECOND, DATEADD(SECOND, lmcj_timestamp, '1970-01-01'), GETUTCDATE())) > 2592000)
                    '''
        rs_lm=ms.ExecQuerydict(sql_lm)
        task_len=1
    elif taskid==13:
        sql_pro=''' select id,url,itemid from shopeephadpro where shop_location='Mainland China' and is_cj=0 order by month_sold desc'''
        rs_lm=ms.ExecQuery(sql_pro)
        task_len=10
    elif taskid==14:
        sql_lm='''  select * from phlmurl where (ysdpurl is not null and  ysdpurl not in ('','原始链接下架')) 
                    and (dpcj_timestamp is null or ABS(DATEDIFF(SECOND, DATEADD(SECOND, dpcj_timestamp, '1970-01-01'), GETUTCDATE())) > 2592000)
                '''
        rs_lm=ms.ExecQuerydict(sql_lm)
        task_len=1
    elif taskid==15:

        aff_zz=ms.ExecNoQuery('update keyword_1688 set is_cj=0 where is_cj=1 and gender<3 and DATEDIFF(DAY, cj_time, GETDATE()) >=9999')
        logging.info(f'任务《{rw_name}》,重置 {aff_zz} 个关键词采集状态')

        sql_lm='''WITH agg AS (
                    SELECT 
                        keyword,
                        '[' + STRING_AGG(QUOTENAME(emarket,'"'), ',') + ']' AS emarket_json,
                        MAX(keyword_type) AS keyword_type,
                        min(is_cj) AS max_is_cj,
                        MIN(gender) AS min_gender
                    FROM keyword_1688
                    WHERE language='中文'
                    AND (gender<4 or (gender=4 and size_weight_type in ('体积大,重量小','体积小,重量大') ))
                    GROUP BY keyword
                )
                SELECT top 1000 keyword, emarket_json, keyword_type,min_gender
                FROM agg p
                WHERE max_is_cj = 0
                
                AND NOT EXISTS (
                    SELECT 1
                    FROM Blacklist b
                    WHERE b.word_language = '中文'
                    AND p.keyword LIKE '%' + b.word_blacklist + '%'
                ) order by min_gender
                ;
            '''
        rs_lm=ms.ExecQuerydict(sql_lm)
        task_len=10
    
    elif taskid==16:
        sql_pro0='''
                    select id, url, itemid from S1688Pro p
                        where  (is_cj = 0 OR (is_cj = 1 AND attributes IS NULL)) 
                        AND NOT EXISTS (
                            SELECT 1
                            FROM Blacklist b
                            WHERE b.word_language = '中文'
                            AND p.pname LIKE '%' + b.word_blacklist + '%'
                        )
                        order by p.keword_gender
            '''
    
        rs_lm=ms.ExecQuery(sql_pro0)
        task_len=80

    elif taskid==25:

        sql_lm= '''select category_id,tag_name,cj_ind from  ShopeeKOLCategory where cj_ind<46656 '''
        rs_lm=ms.ExecQuery(sql_lm)
        task_len=1

    elif taskid==36:
        sql_pro0='''
                select distinct Offerid from S1688Order o where 
                NOT EXISTS (
                    SELECT 1 FROM s1688pro s
                    WHERE s.itemid= o.Offerid
                )
            '''
    
        rs_lm=ms.ExecQuery(sql_pro0)
        task_len=80

    else: 
        ms.ExecNoQuery('update [task] set state=0,tcount=0,LastHouTaiID=null where [id]=?',(taskid,))
        change_task_status(taskid,0)
        return
    


    if len(rs_lm)==0:
        logging.info(f'任务《{rw_name}》,暂时不需要采集...')        
        ms.ExecNoQuery('update [task] set state=0,tcount=0,LastHouTaiID=null where [id]=?',(taskid,))
        change_task_status(taskid,0)
        return

    task_count=len(rs_dp)
    logging.info(f'任务《{rw_name}》,共{task_count} 个买家号可以用来采集,需采集任务总数:{len(rs_lm)}')

    if isMulti==0 or isMulti==-1:
        djj=0
        for r in rs_dp:
            bid=r['BrowserID']
            bname=r['DpName']
            if taskid in (25,):
                cur_shop=Shop(r)
            else:
                cur_shop=PHShop(r)
            task_list=rs_lm[djj*task_len:(djj+1)*task_len]
            if not task_list:
                break
            logging.info(f'买家号:《{bname}》,分配到:{len(task_list)} 个采集任务,开始采集...')
            if taskid in [12,14,15,16,25]:
                cur_shop.lmitem_list=task_list
            elif taskid==13:
                cur_shop.pro_list=task_list
            elif taskid==36:
                cur_shop.offerid_list=task_list
            try:
                cur_shop.start_task(taskid)
            except Exception as e:
                traceback.print_exc()
                time.sleep(3)
            task_count-=1
            logging.info(f'{rw_name},剩买家号数:{task_count}\n\n')
            djj+=1
            ms.ExecNoQuery('update [task] set [tcount]=[tcount]-1 where [id]=?',(taskid,))
    
    else:
        print(f'{rw_name},多进程运行,进程数:{isMulti}')
        manager = Manager()
        queue = manager.Queue()
            # 多进程池的初始化

        with Pool(processes=isMulti) as pool:  # 根据实际情况设置并发进程数量
            results = []
            djj=0
            for r in rs_dp:
                
                hid=r['id']
                if taskid in (25,):
                    cur_shop=Shop(r)
                else:
                    cur_shop=PHShop(r)
                task_list=rs_lm[djj*task_len:(djj+1)*task_len]
                if not task_list:
                    break
                if taskid in[12,14,15,16,25]:
                    cur_shop.lmitem_list=task_list
                elif taskid==13:
                    cur_shop.pro_list=task_list
                elif taskid==36:
                    cur_shop.offerid_list=task_list
                result = pool.apply_async(
                    process_shop, args=(taskid,cur_shop,queue),
                    callback=lambda _,hid=hid: on_task_complete(ms,queue,task_count,taskid,hid)
                )
                results.append(result)
                djj+=1

            # 监控任务执行情况
            pool.close()
            for result in results:
                result.wait()  # 等待每个进程完成，无超时限制

            # 所有任务执行完成后继续
            pool.join()


    ms.ExecNoQuery('update [task] set state=0,tcount=0,LastHouTaiID=null where [id]=?',(taskid,))
    change_task_status(taskid,0)


if __name__ == "__main__":
    main()